package cn.ac.iscas.service.executor.dameng;

import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.config.DamengConfig;
import cn.ac.iscas.dao.IJobMetadataDao;
import cn.ac.iscas.domain.job.GlobeJobPeriodMap;
import cn.ac.iscas.domain.job.period.JobExecutePeriod;
import cn.ac.iscas.service.log.DamengLogService;
import cn.ac.iscas.service.queue.SqlExecuteQueueService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.Collectors;

import static java.lang.Thread.sleep;

@Service
public class DamengJobExecutor {

    @Autowired
    private DamengConfig damengConfig;

    @Autowired
    private DamengLogService damengLogService;

    @Autowired
    private IJobMetadataDao jobMetadataDao;

    public void execute(String jobId) {
        JobExecutePeriod jobExecutePeriod = GlobeJobPeriodMap.get(jobId);
        if (jobExecutePeriod == null) {
            executeOnce(jobId);
        } else {
            scheduleExecution(jobId, jobExecutePeriod.getDelay(), jobExecutePeriod.getPeriod());
        }

    }

    public void executeOnce(String jobId) {
        damengLogService.send(jobId, String.format("%s 开始执行任务[%s]: ", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), jobId));
        System.out.println("Current time: " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Date.from(ZonedDateTime.now(ZoneId.of("Asia/Shanghai")).toInstant())));
        SqlExecuteQueueService sqlExecuteQueueService = SpringUtil.getBean(SqlExecuteQueueService.class);
        ConcurrentLinkedQueue<String> sqls = sqlExecuteQueueService.getSqlsByJobId(jobId);
        List<String> sortedSqls = sqls.stream()
                .sorted(Comparator.comparing(s -> {
                    if (s.startsWith("CREATE SCHEMA")) {
                        return 0;
                    } else if (s.startsWith("CREATE GLOBAL TEMPORARY")) {
                        return 1;
                    } else {
                        return 2;
                    }
                }))
                .collect(Collectors.toList());

        String damengJdbcUrl = String.format("jdbc:dm://%s:%s", damengConfig.getHost(), damengConfig.getPort());

        Connection connection = null;
        Statement statement = null;

        try {
            connection = DriverManager.getConnection(damengJdbcUrl, damengConfig.getUser(), damengConfig.getPassword());
            statement = connection.createStatement();
            connection.setAutoCommit(false);

            for (String sql : sortedSqls) {
                statement.addBatch(sql);
                System.out.println(sql);
            }

            int[] result = statement.executeBatch();
            connection.commit();

            damengLogService.send(jobId, String.format("%s [FINISHED]已成功执行[%s]任务", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), jobId));

            System.out.println("Successfully executed " + result.length + " SQL statements.");
            jobMetadataDao.updateJobStatus(jobId, "SUCCEEDED");
        } catch (SQLException e) {
            System.err.println("SQL execution failed: " + e.getMessage());
            jobMetadataDao.updateJobStatus(jobId, "PAUSE");
        } finally {
            String dropTemporarySchema = String.format("DROP SCHEMA IF EXISTS \"%s\" CASCADE", jobId);
            try {
                statement.execute(dropTemporarySchema);
                connection.commit();
            } catch (SQLException e) {
                throw new RuntimeException(e);
            }

            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

        }
    }


    public void scheduleExecution(String jobId, long delay, long period) {
        Timer timer = new Timer();
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                executeOnce(jobId);
            }
        };

        timer.schedule(task, delay, period);
    }


}
